package study.flink.tcauction;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.UUID;

public class TcAuctionItemJob {

    public static void main(String[] args) throws Exception {
        //up 1 0 10 0
        //buy 1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        OutputTag<GoodsEvent> downOutput = new OutputTag<GoodsEvent>("downOutput") {
        };
//        OutputTag<GoodsEvent> intervalOutput = new OutputTag<>("intervalOutput");
        OutputTag<GoodsEvent> newOuput = new OutputTag<GoodsEvent>("newOuput") {
        };
        SocketTextStreamFunction source = new SocketTextStreamFunction("vm-ubuntu", 8888, "\n", 1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        SingleOutputStreamOperator<GoodsEvent> dataStream =
//                env.readTextFile("d:/flinkinput.txt")
                env.addSource(source)
                        .flatMap(new FlatMapFunction<String, GoodsEvent>() {
                            @Override
                            public void flatMap(String value, Collector<GoodsEvent> out) throws Exception {
                                String[] values = value.split("\r\n");
                                for (String s : values) {
                                    if (s == null || s.isEmpty()) {
                                        continue;
                                    }
                                    if (s.startsWith("up")) {
                                        String[] fields = s.split(" ");
                                        GoodsEvent event = new GoodsEvent();
                                        event.setEvent(fields[0]);
                                        event.setGoodsId(Integer.parseInt(fields[1]));
                                        event.setType(Integer.parseInt(fields[2]));
                                        event.setNumber(Integer.parseInt(fields[3]));
                                        event.setInterval(Integer.parseInt(fields[4]));
                                        out.collect(event);
                                    } else if (s.startsWith("buy")) {
                                        String[] fields = s.split(" ");
                                        GoodsEvent event = new GoodsEvent();
                                        event.setEvent(fields[0]);
                                        event.setGoodsId(Integer.parseInt(fields[1]));
                                        out.collect(event);
                                    }
                                }
                            }
                        }).keyBy(GoodsEvent::getGoodsId)
                        .map(new GoodsEventMapper())
                        .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<GoodsEvent>() {
                            @Override
                            public long extractAscendingTimestamp(GoodsEvent element) {
                                return System.currentTimeMillis();
                            }
                        })
                        .keyBy(e -> e.getItemId())
                        .process(new KeyedProcessFunction<String, GoodsEvent, GoodsEvent>() {

                            ValueState<GoodsEvent> state;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                super.open(parameters);
                                state = getRuntimeContext().getState(new ValueStateDescriptor<>("goods-process", GoodsEvent.class));
                            }

                            @Override
                            public void processElement(GoodsEvent value, Context ctx, Collector<GoodsEvent> out) throws Exception {
                                if ("down".equals(value.getEvent())) {
                                    ctx.output(downOutput, value);
                                } else if ("new".equals(value.getEvent())) {
                                    ctx.output(newOuput, value);
                                } else if ("interval".equals(value.getEvent())) {
                                    state.update(value);
                                    ctx.timerService().registerProcessingTimeTimer(ctx.timestamp() + value.getInterval() * 1000);
                                }
                            }

                            @Override
                            public void onTimer(long timestamp, OnTimerContext ctx, Collector<GoodsEvent> out) throws Exception {
                                super.onTimer(timestamp, ctx, out);
//                        out.collect(state.value());
                                ctx.output(newOuput, state.value());
                                state.clear();
                            }
                        });
        dataStream.getSideOutput(newOuput).addSink(new SinkFunction<GoodsEvent>() {
            @Override
            public void invoke(GoodsEvent value, Context context) throws Exception {
                System.out.println("new " + value.getItemId());
            }
        });
        /*dataStream.getSideOutput(downOutput).addSink(new SinkFunction<GoodsEvent>() {
            @Override
            public void invoke(GoodsEvent value, Context context) throws Exception {
                System.out.println("down");
            }
        });*/
        env.execute();
//        LockSupport.park();
    }

    static class GoodsEventMapper extends RichMapFunction<GoodsEvent, GoodsEvent> {

        ValueState<GoodsState> state;

        @Override
        public GoodsEvent map(GoodsEvent value) throws Exception {
            if (value.getEvent().equals("up")) {
                GoodsState goodsState = new GoodsState();
                goodsState.setGoodsId(value.getGoodsId());
                goodsState.setType(value.getType());
                goodsState.setInterval(value.getInterval());
                goodsState.setNumber(value.getNumber());
                state.update(goodsState);
            }
            GoodsState goodsState = state.value();
            GoodsEvent nextEvent = new GoodsEvent();
            nextEvent.setGoodsId(goodsState.getGoodsId());
            nextEvent.setItemId(UUID.randomUUID().toString());
            if (goodsState.getNumber() > 0) {
                if (goodsState.getInterval() != null) {
                    nextEvent.setEvent("interval");
                    nextEvent.setInterval(goodsState.getInterval());
                } else {
                    nextEvent.setEvent("new");
                }
                goodsState.setNumber(goodsState.getNumber() - 1);
                state.update(goodsState);
            } /*else {
                nextEvent.setEvent("down");
            }*/
            return nextEvent;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            state = getRuntimeContext().getState(new ValueStateDescriptor<>("goods", GoodsState.class));
        }
    }

}
